package com.darcytech.debezium.core.rpc;

import com.darcytech.debezium.common.utils.StringLockUtils;
import com.darcytech.debezium.core.api.Api;
import com.darcytech.debezium.core.api.impl.ApiSender;
import com.darcytech.debezium.core.rpc.proto.Rpc;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RpcClientHandler extends SimpleChannelInboundHandler<Rpc.Resp> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcClientHandler.class);

    private final ApiSender handleApi;

    public RpcClientHandler(ApiSender handleApi) {
        this.handleApi = handleApi;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        SocketChannel channel = (SocketChannel) ctx.channel();
        this.handleApi.setChannel(channel);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.WRITER_IDLE) {
                ctx.writeAndFlush(Api.PING);
            }
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Rpc.Resp resp) {
        Object returnDataLock = StringLockUtils.getStringLock(resp.getRequestId());
        if (returnDataLock == null) {
            LOGGER.warn("wait resp timeout requestId : {}", resp.getRequestId());
            return;
        }
        synchronized (returnDataLock) {
            this.handleApi.setRes(resp.getRequestId(), resp.getData());
            returnDataLock.notify();
        }
    }
}
